package promote.brozen.reactor;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;
import reactor.core.publisher.UnicastProcessor;

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

/**
 * @author Brozen
 * @date 2020/8/10 11:24 AM
 * @email brozen@qq.com
 */
public class ColdAndHot {

    public static void main(String[] args) throws Exception {
        hot();
    }

    /**
     * 默认是cold 冷发布者
     * 这种发布者，每一次有消费者调用subscribe()方法订阅，都会调用同一个publisher的subscribe方法
     * 注意是调用同一个publisher，多次订阅不会多次创建publisher
     *
     * 在这种情况下，每个订阅者都能拿到事件源发布的全部数据
     */
    private static void defaultCold() {

        AtomicInteger subscribeCount = new AtomicInteger(0);
        Flux<String> flux = Flux.from(new Publisher<String>() {
            @Override
            public void subscribe(Subscriber<? super String> s) {
                System.out.println("这一次的publisher:" + this.hashCode());
                subscribeCount.incrementAndGet();
                for (int i = 0; i < 10; i++) {
                    s.onNext(subscribeCount.get() + " 发布 " + i);
                }
            }
        });

        System.out.println("first time subscribe");
        flux.subscribe(System.out::println);

        System.out.println("second time subscribe");
        flux.subscribe(System.out::println);
    }


    /**
     * hot发布者用Processor实现
     * 在第一次被订阅之前，发布者发布的数据会被缓存下来，第一次被订阅后，集中发射给订阅者；之后订阅的订阅者，只能接受到从订阅时刻之后发射的数据
     *
     * 可以设置多个订阅者到齐之后才开始第一次将缓存的数据集中发射给订阅者，ConnectableFlux用意在此
     */
    private static void hot() throws InterruptedException {
        UnicastProcessor<String> hotSource = UnicastProcessor.create();

        AtomicInteger counter = new AtomicInteger(0);
        Timer timer = new Timer();
        timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                hotSource.onNext("Send " + counter.getAndIncrement());
            }
        }, 1000, 1000);

        // 这个ConnectableFlux可以设置n个订阅者订阅后，才开始关联到上游Flux订阅数据
        ConnectableFlux<String> connectableFlux = hotSource.publish();

        Flux<String> flux = connectableFlux.autoConnect();
        System.out.println("first time subscribe");
        flux.subscribe(System.out::println);

        Thread.sleep(3000L);

        System.out.println("second time subscribe");
        flux.subscribe(System.out::println);

        flux.blockLast();
    }

}
